package com.example.source.debezium;

import org.apache.kafka.connect.data.Struct;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.StringJoiner;

/**
 * 通用 SQL 生成器：根据 Debezium 事件拼接 INSERT / UPDATE / DELETE SQL
 */
public class SQLGenerator {

    /**
     * 构造 INSERT SQL
     */
    public static String buildInsertSQL(String table, Struct after) {
        if (after == null) return null;

        Set<String> fields = after.schema().fields().stream().map(f -> f.name()).collect(java.util.stream.Collectors.toSet());

        StringJoiner columns = new StringJoiner(", ");
        StringJoiner values = new StringJoiner(", ");

        for (String field : fields) {
            Object val = after.get(field);
            columns.add("`" + field + "`");
            values.add(toSQLValue(val));
        }

        return "INSERT INTO `" + table + "` (" + columns + ") VALUES (" + values + ");";
    }

    /**
     * 构造 UPDATE SQL
     */
    public static String buildUpdateSQL(String table, Struct before, Struct after) {
        if (before == null || after == null) return null;

        StringJoiner setClause = new StringJoiner(", ");
        StringJoiner whereClause = new StringJoiner(" AND ");

        // 构造 SET
        for (org.apache.kafka.connect.data.Field field : after.schema().fields()) {
            String name = field.name();
            Object beforeVal = before.get(name);
            Object afterVal = after.get(name);

            // 只更新变化的字段
            if ((beforeVal == null && afterVal != null) ||
                    (beforeVal != null && !beforeVal.equals(afterVal))) {
                setClause.add("`" + name + "`=" + toSQLValue(afterVal));
            }
        }

        // 构造 WHERE（用 before 值防止丢失数据）
        for (org.apache.kafka.connect.data.Field field : before.schema().fields()) {
            String name = field.name();
            Object val = before.get(name);
            if (val != null) {
                whereClause.add("`" + name + "`=" + toSQLValue(val));
            }
        }

        return "UPDATE `" + table + "` SET " + setClause + " WHERE " + whereClause + ";";
    }

    /**
     * 构造 DELETE SQL
     */
    public static String buildDeleteSQL(String table, Struct before) {
        if (before == null) return null;

        StringJoiner whereClause = new StringJoiner(" AND ");
        for (org.apache.kafka.connect.data.Field field : before.schema().fields()) {
            String name = field.name();
            Object val = before.get(name);
            if (val != null) {
                whereClause.add("`" + name + "`=" + toSQLValue(val));
            }
        }

        return "DELETE FROM `" + table + "` WHERE " + whereClause + ";";
    }

    /**
     * 把 Java 对象转成 SQL 字符串
     */
    private static String toSQLValue(Object val) {
        if (val == null) {
            return "NULL";
        }
        if (val instanceof Number || val instanceof Boolean) {
            return val.toString();
        }
        return "'" + val.toString().replace("'", "''") + "'";
    }
}
